package com.atguigu.gamll.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.atguigu.gamll.realtime.app.func.DimSink;
import com.atguigu.gamll.realtime.app.func.MyDebeziumDeserializationSchema;
import com.atguigu.gamll.realtime.app.func.TableProcessFunction;
import com.atguigu.gamll.realtime.beans.TableProcess;
import com.atguigu.gamll.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

// 业务数据动态分流
// 需要启动的进程
//      zk、kf、maxwell
public class BaseDBApp extends BaseApp {
    public static void main(String[] args) throws Exception {
        // TODO 1.基本环境准备
        // TODO 2.检查点设置
        BaseDBApp baseDBApp = new BaseDBApp();
        baseDBApp.baseEntry();
    }

    @Override
    public void biz(StreamExecutionEnvironment env) {
        // TODO 3.从kafka消费数据
        // 3.1 声明消费主题和消费者组
        String topic = "ods_base_db_m";
        String groupId = "base_db_app_group";
        // 3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        // 3.3 消费数据  封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaSource);

        // TODO 4.转换数据类型  jsonStr --> jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        // TODO 5.elt
        SingleOutputStreamOperator<JSONObject> filterDS = jsonObjDS.filter(
                new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject) throws Exception {
                        boolean flag = jsonObject.getString("table") != null &&
                                jsonObject.getString("table").length() > 0 &&
                                jsonObject.getJSONObject("data") != null &&
                                jsonObject.getString("data").length() > 3;
                        return flag;
                    }
                }
        );
        //filterDS.print();

        // TODO 6.读取配置表得到配置流
        // 6.1 创建MySQLSource
        DebeziumSourceFunction<String> mysqlSource = MySQLSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall0819_realtime")
                .tableList("gmall0819_realtime.table_process")
                .startupOptions(StartupOptions.initial())
                .username("root")
                .password("000000")
                .deserializer(new MyDebeziumDeserializationSchema())
                .build();
        // 6.2 从mysql中读取数据，得到配置流
        DataStreamSource<String> mysqlStrDS = env.addSource(mysqlSource);

        // TODO 7.广播配置流，形成广播流，定义广播状态
        MapStateDescriptor<String, TableProcess> mapStateDescriptor =
                new MapStateDescriptor<>("mapStateDescriptor", String.class, TableProcess.class);
        BroadcastStream<String> broadcastDS = mysqlStrDS.broadcast(mapStateDescriptor);

        // TODO 8.将主流和广播流关联在一起
        BroadcastConnectedStream<JSONObject, String> connectDS = filterDS.connect(broadcastDS);

        // TODO 9.处理关联后的数据---动态分流  事实-->主流  维度-->侧输出流
        // 定义侧输出流
        OutputTag<JSONObject> dimTag = new OutputTag<JSONObject>("dimTag") {
        };

        SingleOutputStreamOperator<JSONObject> realDS = connectDS.process(new TableProcessFunction(dimTag, mapStateDescriptor));

        DataStream<JSONObject> dimDS = realDS.getSideOutput(dimTag);
        realDS.print(">>>");
        dimDS.print("~~~");

        // TODO 10.将维度侧输出流数据写到phoenix中
        dimDS.addSink(new DimSink());

        // TODO 11.将主流事实数据写到kafka主题中
        realDS.addSink(MyKafkaUtil.getKafkaSinkSchema(new KafkaSerializationSchema<JSONObject>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObject, @Nullable Long aLong) {
                String topic = jsonObject.getString("sink_table");
                return new ProducerRecord<byte[], byte[]>(topic, jsonObject.getJSONObject("data").toJSONString().getBytes());
            }
        }));
    }
}
